package com.patsnap.data.npd.dw.etl.job.hotfix;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.patsnap.data.npd.dw.core.pipeline.journal.DwdJournalCalculationPipeline;
import com.patsnap.data.npd.dw.etl.job.AbstractSparkBatchJob;
import com.patsnap.data.npd.dw.etl.utils.SparkUtils;
import com.patsnap.data.npd.dw.repository.DBOperator;
import com.patsnap.data.npd.dw.schema.ods.journal.OdsJournalJBestInfo;
import com.patsnap.one.etl.cdc.CdcRecord;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.cli.CommandLine;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.util.LongAccumulator;

@Slf4j
public class HotfixJournalJob extends AbstractSparkBatchJob {

    public static void main(String[] args) throws Exception {
        if (args.length == 0) {
            args = new String[]{
                    "--input", "/Users/yingxiuyong/Desktop/spark_test",
                    "--apollo.app.cluster",
                    "suzhou_office",
                    "--apollo.app.profile",
                    "patent_batch",
                    "--apollo.app.id",
                    "s-data-patent-dw-pipeline"};
        }
        new HotfixJournalJob().laucher(args);
    }

    @Override
    protected void process(SparkSession spark, CommandLine cmd) {
        String input = cmd.getOptionValue("input");
        LongAccumulator total = spark.sparkContext().longAccumulator("total");
        LongAccumulator error = spark.sparkContext().longAccumulator("error");
        //spark.read().textFile(input).repartition(50)
        spark.read().parquet(input).repartition(50).toJSON()
                //spark.read().textFile(input).repartition(50)
                .foreachPartition(iterator -> {
                    DBOperator client = initDBOperator(cmd);
                    while (iterator.hasNext()) {
                        String line = iterator.next();
                        try {
                            fixJBestJournalId(client, line);
                            total.add(1);
                        } catch (Exception e) {
                            log.error(line, e);
                            error.add(1);
                        }
                    }
                });
        SparkUtils.printResult(total, error);
    }


    private void fixJBestJournalId(DBOperator client, String line) throws Exception {
        JSONObject jo = JSON.parseObject(line);
        String jBestId = jo.getString("jbest_id");
        OdsJournalJBestInfo original = OdsJournalJBestInfo.builder()
                .jBestId(jBestId)
                .build();
        OdsJournalJBestInfo odsJournalJBestInfo = client.getOdsJournalRepo().queryOneByPrimaryKey(original, OdsJournalJBestInfo.class);
        CdcRecord build = CdcRecord.builder()
                .source(CdcRecord.Source.builder().table("ods_journal_jbest_info").build())
                .after(JSON.parseObject(JSON.toJSONString(odsJournalJBestInfo)))
                .build();

        DwdJournalCalculationPipeline pipeline = new DwdJournalCalculationPipeline();
        DwdJournalCalculationPipeline.DwdJournalCalculationContext context = pipeline.initContext(build);
        pipeline.extract(client, context);
        pipeline.load(client, context);
    }
}
